package com.behavioranalysis.flinkprogram.flink.session;

import com.behavioranalysis.flinkprogram.constant.Constants;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.flink.api.common.accumulators.Accumulator;

import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 用例题中的方法
 * <p>
 * Similar to Accumulator, but the type of items to add and the result value
 * must be the same.
 */
public class SessionAggrStatAccumulator implements Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> {

    private static final long serialVersionUID = 1L;

    /**
     * Constants.SESSION_COUNT + "=0|"
     * + Constants.TIME_PERIOD_1s_3s + "=0|"
     * + Constants.TIME_PERIOD_4s_6s + "=0|"
     * + Constants.TIME_PERIOD_7s_9s + "=0|"
     * + Constants.TIME_PERIOD_10s_30s + "=0|"
     * + Constants.TIME_PERIOD_30s_60s + "=0|"
     * + Constants.TIME_PERIOD_1m_3m + "=0|"
     * + Constants.TIME_PERIOD_3m_10m + "=0|"
     * + Constants.TIME_PERIOD_10m_30m + "=0|"
     * + Constants.TIME_PERIOD_30m + "=0|"
     * + Constants.STEP_PERIOD_1_3 + "=0|"
     * + Constants.STEP_PERIOD_4_6 + "=0|"
     * + Constants.STEP_PERIOD_7_9 + "=0|"
     * + Constants.STEP_PERIOD_10_30 + "=0|"
     * + Constants.STEP_PERIOD_30_60 + "=0|"
     * + Constants.STEP_PERIOD_60 + "=0";
     */

    private ConcurrentSkipListMap<String, MutableLong> values = new ConcurrentSkipListMap<>();

    public SessionAggrStatAccumulator() {
        values.put(Constants.SESSION_COUNT, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_1s_3s, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_4s_6s, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_7s_9s, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_10s_30s, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_30s_60s, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_1m_3m, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_3m_10m, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_10m_30m, new MutableLong(0));
        values.put(Constants.TIME_PERIOD_30m, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_1_3, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_4_6, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_7_9, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_10_30, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_30_60, new MutableLong(0));
        values.put(Constants.STEP_PERIOD_60, new MutableLong(0));
    }

    /**
     * 将 value 按照一定的规则添加到 values 中
     *
     * @param value
     */
    @Override
    public void add(String value) {
        if (values.containsKey(value)) {
            values.get(value).increment();
        } else {
            values.put(value, new MutableLong(1));
        }
    }

    @Override
    public ConcurrentSkipListMap<String, MutableLong> getLocalValue() {
        return values;
    }

    @Override
    public void resetLocal() {
        values.clear();
    }

    /**
     * 将不同分区的累加器值 全部聚合起来
     *
     * @param other
     */
    @Override
    public void merge(Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> other) {
        other.getLocalValue().forEach((k, v) -> {
            if (values.containsKey(k)) {
                values.get(k).add(v.longValue());
            } else {
                values.put(k, v);
            }
        });
    }

    @Override
    public Accumulator<String, ConcurrentSkipListMap<String, MutableLong>> clone() {
        SessionAggrStatAccumulator sessionAggrStatAccumulator = new SessionAggrStatAccumulator();
        sessionAggrStatAccumulator.getLocalValue().putAll(this.values);
        return sessionAggrStatAccumulator;
    }
}
